package com.pw.study.flink.demo

import com.pw.study.flink.entities.WordBeanOne
import com.pw.study.flink.source.CustomWordSource
import org.apache.flink.api.java.io.TextInputFormat
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.FileProcessingMode
import org.apache.flink.streaming.api.scala._

object WordCountByStream {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(1)
    //设置时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val filePath = "data/file/words.txt"
    val format = new TextInputFormat(new Path(filePath))
    val ds: DataStream[String] = env.readFile(format, filePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 1)

    val broadcast: DataStream[WordBeanOne] = env.addSource(new CustomWordSource()).broadcast
    val dsMap: DataStream[WordBeanOne] = ds.flatMap(_.split(" ")).map(x => {
      WordBeanOne(x, 1L)
    })
    broadcast.print("broadcast")
    ds.print("word:")
    dsMap.print("wordOne")


    env.execute("StreamWordCount")
  }
}
